package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdCourseReviewInfo extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdCourseReviewInfo().init(
                3001,
                3,
                "DwdCourseReviewInfo"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        //读取ods_db
        readOdsDb(tEnv,"DwdCourseReviewInfo");

        //读取review_info
        Table result = tEnv.sqlQuery("select " +
                " data['id'] id ," +
                " data['user_id'] user_id ," +
                " data['course_id'] course_id ," +
                " data['review_txt'] review_txt ," +
                " data['review_stars'] review_stars ," +
                " data['create_time'] create_time ," +
                " data['deleted'] deleted ," +
                " ts ts" +
                " from ods_db" +
                " where `database`='edu' " +
                " and `table`='review_info' " +
                " and `type`='insert'");
        tEnv.createTemporaryView("reviewInfo",result);

        //定义动态表输出topic关联
        tEnv.executeSql("create table dwd_course_review_info(" +
                " id string," +
                " user_id string," +
                " course_id string," +
                " review_txt string," +
                " review_stars string," +
                " create_time string," +
                " deleted string," +
                " ts BIGINT" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_COURSE_REVIEW_INFO)

        );

        result.executeInsert("dwd_course_review_info");

    }
}
